#pragma once

#include "Requester.hpp"
#include "RpcCaller.hpp"
#include "RpcRegistry.hpp"
#include "RpcTopic.hpp"
#include "../common/Dispatcher.hpp"

namespace rpc
{
    namespace client
    {
        class RegistryClient
        {
        public:
            using ptr = std::shared_ptr<RegistryClient>;
            RegistryClient(const std::string &ip, const uint16_t &port)
                : _requester(std::make_shared<Requester>())
                , _provider(std::make_shared<client::ServiceProvider>(_requester))
                , _dispatcher(std::make_shared<Dispatcher>())
            {
                // 设置响应回调函数
                auto rsp_cb = std::bind(&client::Requester::onResponse, _requester.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MSG_TYPE::RSP_SERVICE, rsp_cb);

                // 设置消息回调函数
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                    std::placeholders::_1, std::placeholders::_2);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallback(msg_cb);
                _client->connect();
            }

            bool registryService(const Address &host, const std::string &service)
            {
                return _provider->registryService(_client->connection(), host, service);
            }

        private:
            Requester::ptr _requester;
            client::ServiceProvider::ptr _provider;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        class DiscoveryClient
        {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;
            DiscoveryClient(const std::string &ip, const uint16_t &port, const ServiceDiscoverer::offline_cb &cb)
                : _requester(std::make_shared<Requester>())
                , _discoverer(std::make_shared<ServiceDiscoverer>(_requester, cb))
                , _dispatcher(std::make_shared<Dispatcher>())
            {
                // 设置 处理服务上下线请求的回调函数
                auto rsp_cb = std::bind(&Requester::onResponse, _requester.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MSG_TYPE::RSP_SERVICE, rsp_cb);

                auto req_cb = std::bind(&client::ServiceDiscoverer::onServiceRequest, _discoverer.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<ServiceRequest>(MSG_TYPE::REQ_SERVICE, req_cb);

                // 设置消息回调函数
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                    std::placeholders::_1, std::placeholders::_2);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallback(msg_cb);
                _client->connect();
            }

            bool serviceDiscovery(const std::string &service, Address &host)
            {
                return _discoverer->serviceDiscovery(_client->connection(), service, host);
            }

        private:
            Requester::ptr _requester;
            client::ServiceDiscoverer::ptr _discoverer;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };
        
        
        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            RpcClient(bool enablediscovery, const std::string &ip, const uint16_t &port)
                : _enablediscovery(enablediscovery)
                , _requester(std::make_shared<Requester>())
                , _rpc_caller(std::make_shared<RpcCaller>(_requester))
                , _dispatcher(std::make_shared<Dispatcher>())
            {
                // 设置 处理rpc响应的回调函数
                auto rsp_cb = std::bind(&Requester::onResponse, _requester.get(), 
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MSG_TYPE::RSP_RPC, rsp_cb);

                if (_enablediscovery)
                {
                    auto offline_cb = std::bind(RpcClient::delClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else
                {
                    auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                        std::placeholders::_1, std::placeholders::_2);
                    _rpc_client = ClientFactory::create(ip, port);
                    _rpc_client->setMessageCallback(msg_cb);
                    _rpc_client->connect();
                }
            }

            bool call(const std::string &service, const Json::Value &params, Json::Value &result)
            {
                BaseClient::ptr client = getClient(service);
                if (client.get() == nullptr)
                {
                    return false;
                }
                // 通过客户端连接发送rpc请求
                return _rpc_caller->call(client->connection(), service, params, result);
            }
            
            bool call(const std::string &service, const Json::Value &params, RpcCaller::JsonAsyncResponse &resp)
            {
                BaseClient::ptr client = getClient(service);
                if (client.get() == nullptr)
                {
                    return false;
                }
                return _rpc_caller->call(client->connection(), service, params, resp);
            }

            bool call(const std::string &service, const Json::Value &params, const RpcCaller::JsonResponseCallback &cb)
            {
                BaseClient::ptr client = getClient(service);
                if (client.get() == nullptr)
                {
                    return false;
                }
                return _rpc_caller->call(client->connection(), service, params, cb);
            }

        private:
            BaseClient::ptr getClient(const std::string &service)
            {
                // std::unique_lock<std::mutex> lock(_mutex); // 加不加锁取决于，是否会有资源访问冲突
                if (_discovery_client)
                {
                    Address host;
                    bool ret = _discovery_client->serviceDiscovery(service, host);
                    if (ret == false)
                    {
                        ELOG("not found the [%s] service", service.c_str());
                        return BaseClient::ptr();
                    }
                    // 有现成的就使用，没有就创建
                    BaseClient::ptr client = getClient(host);
                    if (client.get() == nullptr)
                    {
                        client = newClient(host);
                    }
                    return client;
                }
                else 
                {
                    return _rpc_client;
                }
            }

            BaseClient::ptr getClient(const Address &host) // 在启用服务发现下，通过地址信息找对应客户端
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _host_map_client.find(host);
                if (it == _host_map_client.end())
                {
                    return BaseClient::ptr();
                }
                return it->second;
            }

            BaseClient::ptr newClient(const Address &host)
            {
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), 
                    std::placeholders::_1, std::placeholders::_2);
                BaseClient::ptr client = ClientFactory::create(host.first, host.second);
                client->setMessageCallback(msg_cb);
                client->connect();
                
                putClient(host, client);
                return client;
            }

            void putClient(const Address &host, const BaseClient::ptr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _host_map_client[host] = client;
            }

            void delClient(const Address &host) // 服务提供者下线时，在本地删除对应主机和客户端信息
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _host_map_client.erase(host);
            }

            struct AddressHash
            {
                size_t operator()(const Address &host)
                {
                    std::string addr = host.first + std::to_string(host.second);
                    return std::hash<std::string>()(addr);
                }
            };

        private:
            std::mutex _mutex;
            Requester::ptr _requester;
            RpcCaller::ptr _rpc_caller;
            Dispatcher::ptr _dispatcher;

            bool _enablediscovery; // 是否启用服务发现
            DiscoveryClient::ptr _discovery_client;
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _host_map_client; // <127.0.0.1:8080, client>
            BaseClient::ptr _rpc_client; // 用于未启用服务发现客户端
        };


        class TopicClient
        {
        public:
            using ptr = std::shared_ptr<TopicClient>;
            TopicClient(const std::string ip, const uint16_t &port)
                : _requester(std::make_shared<Requester>())
                , _topic_manager(std::make_shared<TopicManager>(_requester))
                , _dispatcher(std::make_shared<Dispatcher>())
            {
                // 设置回调函数
                auto msg_rsp = std::bind(&Requester::onResponse, _requester.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MSG_TYPE::RSP_TOPIC, msg_rsp);

                auto msg_req = std::bind(&TopicManager::onPublish, _topic_manager.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<TopicRequest>(MSG_TYPE::REQ_TOPIC, msg_req);

                _client = ClientFactory::create(ip, port);
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                    std::placeholders::_1, std::placeholders::_2);

                _client->setMessageCallback(msg_cb);
                _client->connect();
            }

            bool create(const std::string &topic_key)
            {
                return _topic_manager->topicCreate(_client->connection(), topic_key);
            }

            bool remove(const std::string &topic_key)
            {
                return _topic_manager->topicRemove(_client->connection(), topic_key);
            }

            bool subscribe(const std::string &topic_key, const TopicManager::SubsCallback &cb)
            {
                return _topic_manager->topicSubscrive(_client->connection(), topic_key, cb);
            }

            bool unsubscribe(const std::string &topic_key)
            {
                return _topic_manager->topicUnsubscribe(_client->connection(), topic_key);
            }

            bool publish(const std::string &topic_key, const std::string &msg)
            {
                return _topic_manager->topicPublish(_client->connection(), topic_key, msg);
            }

            void shutdown()
            {
                _client->shutdown();
            }

        private:
            Requester::ptr _requester;
            TopicManager::ptr _topic_manager;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };
    }
}